package com.lagou.rpc.consumer.balancer;

import com.lagou.rpc.pojo.Server;
import com.lagou.rpc.pojo.ServerDelayed;
import com.lagou.rpc.pojo.ServerPath;
import lombok.extern.slf4j.Slf4j;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * zookeeper负载均衡器
 */
@Slf4j
@Component
public class ZkLoadBalancer {

    private int index=0;

    @Autowired
    ZkClient zkClient;

    //服务服务路径
    private String serverList = "/server/list/%s";
    private String serverPath = "/server/list";
    //服务器列表
    private final List<ServerPath> servers = Collections.synchronizedList(new ArrayList<>());
    //待清理的延迟队列
    private final DelayQueue<ServerDelayed> serverDelayedDelayQueue = new DelayQueue<>();


    // 通过静态方法创建ScheduledExecutorService的实例
    private static ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(4);

    //上报间隔单位秒
    private long reportInterval=5;

    //清理间隔,单位秒
    private long cleanInterval=5;

    /**
     * 通过轮询策略获得一个可用的服务
     * @return
     */
    public synchronized Server getServerConfig(){
        servers.sort(ServerPath::compareTo);
        System.out.println(servers.stream().map(p->p.getServer()).collect(Collectors.toList()));
        return servers.get(0).getServer();
    }


    /**
     * 启动负载均衡器
     */
    public void start(){
        //1.获取服务列表
        syncServer();
        //2.添加监听
        zkClient.subscribeChildChanges(serverPath, (a,b)->syncServer());
        //3.每5秒钟进行一次上报
        scheduledExecutorService.scheduleAtFixedRate(this::reportZk,0,reportInterval, TimeUnit.SECONDS);
        //4.开启清理线程
        new Thread(()->doClean()).start();
    }

    /**
     * 进行重新加载服务端列表
     */
    public void syncServer(){
        List<String> children = zkClient.getChildren(serverPath);
        List<ServerPath> collect = children.stream()
                .map(s -> String.format(serverList, s))
                .map(s -> new ServerPath((Server) zkClient.readData(s), s))
                .collect(Collectors.toList());
        for (ServerPath serverPath : collect) {
            //如果不存在进行添加
            if(!servers.contains(serverPath)){
                servers.add(serverPath);
               //添加数据监听
               subscribeDataChanges(serverPath);
            }
        }
        Iterator<ServerPath> itr = servers.iterator();
        while (itr.hasNext()){
            //如果已经不存在了,进行移除
            if(!collect.contains(itr.next())){
                itr.remove();
            }
        }
        System.out.println("当前服务列表"+new ArrayList<>(servers));

    }

    private void subscribeDataChanges(ServerPath serverPath) {
        zkClient.subscribeDataChanges(serverPath.getPath(), new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {
                Server newServer = (Server) data;
                //如果服务器访问时间晚于本地时间的话进行更新
                if(newServer.getDate()!=null&&
                        (serverPath.getServer().getDate()==null||
                                newServer.getDate().after(serverPath.getServer().getDate()))){
                    System.out.println("服务器数据变更old:"+serverPath.getServer()+",new:"+newServer);
                    serverPath.setServer(newServer);
                    timingClean(serverPath);
                }

            }

            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                servers.remove(serverPath);
            }
        });
    }

    /**
     * 向zk进行上报
     */
    public void reportZk(){
        for (ServerPath server : servers) {
            zkClient.writeData(server.getPath(),server.getServer());
            timingClean(server);
        }
    }

    /**
     * 进行定时清理
     * @param server
     */
    public void timingClean(ServerPath server){
        if(server.getServer().getTime()!=null){
            boolean remove = serverDelayedDelayQueue.remove(new ServerDelayed(cleanInterval, server));
            serverDelayedDelayQueue.offer(new ServerDelayed(cleanInterval,server));
            System.out.println("队列总长度:"+serverDelayedDelayQueue.size());
        }
    }

    /**
     * 进行清理
     */
    private void doClean(){
        while (true){
            try {
                Server server = serverDelayedDelayQueue.take().getServerPath().getServer();
                log.info("执清理:");
                System.out.println(server);
                server.cleanTime();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }





}
